#include "MsgInstanceServer.h"


CMsgInstanceServer :: CMsgInstanceServer()
{

   m_eventList.clear();
   m_clientinfo.clear();
   m_serverfd = -1;
   m_Communi = new CCommunication();
   m_eventloop = new CEventLoop();
   m_Communi->SocketIntanceTCP(SOCKET_SERVER, NULL, MSG_SERVER_PORT);
   m_serverfd = m_Communi->SocketFdGet();
   MsgCreateTCPServer();
}


CMsgInstanceServer :: ~CMsgInstanceServer()
{

    for(auto it = m_eventList.begin(); it != m_eventList.end(); it++)
     {
        m_Communi->SocketClose(m_serverfd); 
        // for(auto iti = it->second.begin();iti != it->second.end(); iti++)
        // {
        //        free(iti->second);
        //        iti->second = NULL;
        // }
                
     }
     m_Communi->SocketClose(m_serverfd);
     delete m_Communi;
     delete m_eventloop;
     m_Communi = NULL;
     m_eventloop = NULL;
}

void CMsgInstanceServer :: MsgServerSchedule()
{
    time_t curtime = time(NULL);
    m_eventloop->EventSchedule(20);
    //判断心跳包的差值是否大于两倍个心跳包间隔，是则关闭对应连接
    for(auto it = m_clientinfo.begin(); it != m_clientinfo.end(); )
    {
           if(curtime - it->second.lastkeepalivetime > KEEPALIVESENCOND*2)
           {
                 m_eventloop->EventUnRegister(it->first);
                 m_Communi->SocketClose(it->first);  
                 COMLOG_ERROR_LOG("client %d down\n", it->first);
                 for(auto iti = m_eventList.begin(); iti != m_eventList.end(); iti++)
                    {
                        if(iti->second.find(it->first) != iti->second.end())
                            iti->second.erase(iti->second.find(it->first));
                    }
                 m_clientinfo.erase(it++);   
           }
           else
                 it++;
    }

}


int CMsgInstanceServer :: MsgCreateTCPServer()
{
     m_Communi->SocketListen(100);
     m_eventloop->EventRegister(m_serverfd, this, &CMsgInstanceServer::MsgAcceptClietConnectCB, NULL);
     return 0;
}


void CMsgInstanceServer :: MsgAcceptClietConnectCB(int fd, void *data)
{
     struct sockaddr_in faraddr = {0};
     MsgServerClientInfo msgClientinfo = {0};
     int farfd = -1;
     if((farfd = m_Communi->SocketAccept(&faraddr)) >= 0)
     {
           m_eventloop->EventRegister(farfd, this, &CMsgInstanceServer::MsgReadClietCB, NULL);
           msgClientinfo.clientfd = farfd;
           m_Communi->SocketGetAddrAndPort(&faraddr, msgClientinfo.clientaddr, sizeof(msgClientinfo.clientaddr), msgClientinfo.clientport);
           msgClientinfo.lastkeepalivetime = time(NULL);
           m_clientinfo[farfd] = msgClientinfo;
           COMLOG_ERROR_LOG("client %d connect successed %s %d\n", farfd, msgClientinfo.clientaddr,  msgClientinfo.clientport);
     }
     else
           COMLOG_ERROR_LOG("server accpet failed\n");  
}

MsgPacket * CMsgInstanceServer :: MsgConvertByteToMsgpacket(char *data, int datalen)
{
     MsgPacket *tempMsg = (MsgPacket *)malloc(datalen);
     memcpy(tempMsg, data, datalen);
     return tempMsg;
}


void CMsgInstanceServer :: MsgReadClietCB(int fd, void *data)
{
    int ret = 0;
    char *m_readbuf = (char *)m_clientinfo[fd].readbuf;
    int &m_validlen = m_clientinfo[fd].validlen;
    int m_fd = fd;
    int datalen = 0;
    MsgPacket *tempMsg = NULL;
      //读取网络数据,并进行解析交互包
    if((ret = m_Communi->SocketRead(m_fd, m_readbuf + m_validlen, sizeof(m_clientinfo[fd].readbuf), 30)) > 0)
    {
    
      m_validlen += ret;
      while(m_validlen > 0)
      {
            if(m_validlen >= MSG_PACK_LEN + 2)
            {
                //解包
                if(0 == memcmp(m_readbuf, "HEAD", 4)) 
                {
                    COMLOG_DEBUG_HEXDUMP(m_readbuf, m_validlen);
                    datalen = CONVERT_CAHR_TO_INT(m_readbuf[4], m_readbuf[5], m_readbuf[6], m_readbuf[7]);
                    COMLOG_DEBUG_LOG("find pack head and data len is %d\n", datalen);
                    if(m_validlen < datalen + MSG_PACK_LEN + 2)
                    {
                        COMLOG_DEBUG_LOG("data is not enough\n");
                        return;
                    }          
                    else if(m_readbuf[datalen + MSG_PACK_LEN] != (char)0xaa || m_readbuf[datalen + MSG_PACK_LEN + 1] != (char)0x55)
                    {
                        COMLOG_DEBUG_LOG("end check is error %d %02x %02x\n", MSG_PACK_LEN, m_readbuf[datalen + MSG_PACK_LEN], m_readbuf[datalen + MSG_PACK_LEN + 1]);
                        memmove(m_readbuf , m_readbuf + datalen + MSG_PACK_LEN + 2, m_validlen - datalen - MSG_PACK_LEN - 2);
                        m_validlen -= (datalen + MSG_PACK_LEN + 2);
                        return;
                    }
                }
                else
                {
                    COMLOG_ERROR_LOG("no find pack head\n");
                    memmove(m_readbuf, m_readbuf+4, m_validlen - 4);
                    m_validlen -= 4;   
                    return ;
                }
                //填充结构体
                COMLOG_DEBUG_LOG("start analyse pack data\n");
                if((tempMsg = MsgConvertByteToMsgpacket(m_readbuf, datalen + MSG_PACK_LEN + 2)) != NULL) 
                {
                    COMLOG_DEBUG_LOG("end analyse pack data, start distribute pack data\n");
                    MsgDistributePack(fd, tempMsg);
                    free(tempMsg);
                    tempMsg = NULL;
                }
                //移动数据
                memmove(m_readbuf , m_readbuf + datalen + MSG_PACK_LEN + 2, m_validlen - datalen - MSG_PACK_LEN - 2);
                m_validlen -= (datalen + MSG_PACK_LEN + 2);
            }
            else
            {
                COMLOG_ERROR_LOG("data is not enough, lower pack header\n");
                break;
            }
                

      }
    }
    else if(SOCKETDOWN == ret)
    {
      m_eventloop->EventUnRegister(m_fd);
      m_Communi->SocketClose(m_fd);
      m_clientinfo.erase(m_clientinfo.find(m_fd));
      for(auto it = m_eventList.begin(); it != m_eventList.end(); it++)
      {
          if(it->second.find(m_fd) != it->second.end())
                it->second.erase(it->second.find(m_fd));
      }
    }
    return;
}



void CMsgInstanceServer :: MsgDistributePack(int fd, MsgPacket *msgPacket)
{
    if(NULL == msgPacket)
    {
        COMLOG_ERROR_LOG("msgPacket is NULL\n");
        return ;
    }
       
    int totallen = MSG_PACK_LEN + msgPacket->len + 2;
    auto it = m_eventList.find(msgPacket->event);
    switch(msgPacket->msgtype)
    {
          case PUBLISH: case PUBRESPONSE:
          if(KEEPALIVE_EVENT == msgPacket->event)
          {
              m_clientinfo.find(fd)->second.lastkeepalivetime = time(NULL);
              COMLOG_ERROR_LOG("client %d, keepalive time %d\n", fd, m_clientinfo.find(fd)->second.lastkeepalivetime);
              return;
          }
          if(it == m_eventList.end())
          {
                COMLOG_ERROR_LOG("no find valid event %d\n", msgPacket->event);
                return;
          }
          //开始发送数据
          COMLOG_DEBUG_LOG("send publish data to other client\n");
          for(auto iti = it->second.begin(); iti != it->second.end(); iti++)
          {
              m_Communi->SocketWrite(iti->first, (char *)msgPacket, totallen);
          }
          break;
          case SUBSCRIBE:
          COMLOG_ERROR_LOG("client %d, subscribe event %d\n", fd, msgPacket->event);
          m_eventList[msgPacket->event].insert(pair<int, void *>(fd, NULL));
          break;
          case UNSUBSCRIBE:
          COMLOG_ERROR_LOG("client %d, subscribe event %d\n", fd, msgPacket->event);
          if(m_eventList[msgPacket->event].find(fd) != m_eventList[msgPacket->event].end())
          {
              free(m_eventList[msgPacket->event].find(fd)->second);
              m_eventList[msgPacket->event].find(fd)->second = NULL;
              m_eventList[msgPacket->event].erase(m_eventList[msgPacket->event].find(fd));
          }
          break;
          default: break;
    }
    

}